package com.jiayi;

import org.apache.flink.api.common.functions.FoldFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.wikiedits.WikipediaEditEvent;
import org.apache.flink.streaming.connectors.wikiedits.WikipediaEditsSource;

/**
 * @author: CONCENTRATE
 * @time: 2018/11/20.
 */
public class WikipediaAnalysis {

    public static void main(String[] args) throws Exception {

        //创建一个streaming的上下文
        StreamExecutionEnvironment see = StreamExecutionEnvironment.getExecutionEnvironment();

        //source部分--数据来源部分
        DataStream<WikipediaEditEvent> edits = see.addSource(new WikipediaEditsSource());

        KeyedStream<WikipediaEditEvent, String> keyedEdits = edits
                .keyBy((KeySelector<WikipediaEditEvent, String>) event -> {
                        return event.getUser();
        });

        DataStream<Tuple2<String, Long>> result = keyedEdits
                .timeWindow(Time.seconds(5))//指定窗口宽度
                .fold(new Tuple2<>("", 0L), new FoldFunction<WikipediaEditEvent, Tuple2<String, Long>>() {
                    @Override
                    public Tuple2<String, Long> fold(Tuple2<String, Long> accumulator, WikipediaEditEvent event){
                        accumulator.f0 = event.getUser();
                        accumulator.f1 += event.getByteDiff();
                        return accumulator;
                    }
                });

        result.print();//这里可能发送到kafka中或者落地到其他的存储设备比如redis等等
        see.execute();
    }
}
